#include "redis/store/redis.h"

Redis::Redis() : zsets_(new RedisZSets(options_, cf_handles_)) {}

Redis::~Redis() {
  for (auto handle : cf_handles_) {
    rocksdb::Status s = transcation_db_->DestroyColumnFamilyHandle(handle);
    assert(s.ok());
  }
}

rocksdb::Status Redis::Open(const std::string& db_path,
                            const rocksdb::Options& options) {
  rocksdb::TransactionDB* transcation_db;
  rocksdb::Options ops(options);
  rocksdb::TransactionDBOptions transcation_db_options;
  ops.create_if_missing = true;
  rocksdb::Status s = rocksdb::TransactionDB::Open(ops, transcation_db_options,
                                                   db_path, &transcation_db);
  s = zsets_->Open(transcation_db, s,
                   column_families_);  // distribution index 0 - 2 cf
  if (!s.ok()) {
    return s;
  }

  s = rocksdb::TransactionDB::Open(ops, transcation_db_options, db_path,
                                   column_families_, &cf_handles_,
                                   &transcation_db);
  if (s.ok()) {
    transcation_db_ = std::shared_ptr<rocksdb::TransactionDB>(transcation_db);
    zsets_->SetTranscationDB(transcation_db_);
  }
  return s;
}

rocksdb::Status Redis::CreateColumnFamilies() {
  return transcation_db_->CreateColumnFamilies(column_families_, &cf_handles_);
}

rocksdb::Status Redis::GetProperty(const std::string& property, uint64_t* out) {
}

rocksdb::Status Redis::ScanKeyNum(KeyInfo* key_info) {}

rocksdb::Status Redis::ScanKeys(const std::string& pattern,
                                std::vector<std::string>* keys) {}

rocksdb::Status Redis::Expire(const rocksdb::Slice& key, int32_t ttl) {}

rocksdb::Status Redis::Del(const rocksdb::Slice& key) {}

bool Redis::Scan(const std::string& start_key, const std::string& pattern,
                 std::vector<std::string>* keys, int64_t* count,
                 std::string* next_key) {}

std::shared_ptr<rocksdb::Transaction> Redis::Multi() {
  rocksdb::Transaction* transaction =
      transcation_db_->BeginTransaction(write_options_);
  return std::shared_ptr<rocksdb::Transaction>(transaction);
}

rocksdb::Status Redis::Exec(
    const std::shared_ptr<rocksdb::Transaction>& transaction) {
  return transaction->Commit();
}

rocksdb::Status Redis::ZAdd(
    const rocksdb::Slice& key, const std::vector<ScoreMember>& score_members,
    int32_t* ret, const std::shared_ptr<rocksdb::Transaction>& transaction) {
  return zsets_->ZAdd(key, score_members, ret, transaction);
}

rocksdb::Status Redis::ZRange(
  const rocksdb::Slice& key, int32_t start, int32_t stop,
  std::vector<ScoreMember>* score_members,
  const std::shared_ptr<rocksdb::Transaction>& transaction) {
  return zsets_->ZRange(key, start, stop, score_members, transaction);
}